package io.debezium.connector.postgresql.connection;

import com.huawei.gaussdb.jdbc.jdbc.TimestampUtils;
import com.huawei.gaussdb.jdbc.util.PGInterval;
import io.debezium.annotation.ThreadSafe;
import io.debezium.connector.postgresql.PostgresValueConverter;
import io.debezium.relational.Column;
import io.debezium.relational.DefaultValueConverter;
import io.debezium.relational.ValueConverter;
import io.debezium.util.Collect;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.ZoneOffset;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@ThreadSafe
public class PostgresDefaultValueConverter implements DefaultValueConverter {
    private static Logger LOGGER = LoggerFactory.getLogger(PostgresDefaultValueConverter.class);
    private static final Pattern LITERAL_DEFAULT_PATTERN = Pattern.compile("'(.*?)'");
    private static final Pattern FUNCTION_DEFAULT_PATTERN =
            Pattern.compile("^[(]?[A-Za-z0-9_.]+\\((?:.+(?:, ?.+)*)?\\)");
    private static final Set<String> CURRENT_DATE_TIMES =
            Collect.unmodifiableSet(
                    new String[] {
                        "current_timestamp",
                        "current_time",
                        "current_date",
                        "localtime",
                        "localtimestamp"
                    });
    private static final Set<String> TRIM_DATA_TYPES =
            Collect.unmodifiableSet(
                    new String[] {
                        "bit",
                        "varbit",
                        "bool",
                        "numeric",
                        "float4",
                        "float8",
                        "int2",
                        "int4",
                        "serial",
                        "int8",
                        "bigserial",
                        "smallserial",
                        "uuid",
                        "date",
                        "time",
                        "timestamp",
                        "timestamptz",
                        "interval"
                    });
    private final PostgresValueConverter valueConverters;
    private final Map<String, DefaultValueConverter.DefaultValueMapper> defaultValueMappers;

    public PostgresDefaultValueConverter(
            PostgresValueConverter valueConverters, TimestampUtils timestampUtils) {
        this.valueConverters = valueConverters;
        this.defaultValueMappers =
                Collections.unmodifiableMap(createDefaultValueMappers(timestampUtils));
    }

    public Optional<Object> parseDefaultValue(Column column, String defaultValueExpression) {
        if (defaultValueExpression == null) {
            return Optional.empty();
        } else {
            String dataType = column.typeName();
            DefaultValueConverter.DefaultValueMapper mapper =
                    (DefaultValueConverter.DefaultValueMapper)
                            this.defaultValueMappers.get(dataType);
            if (mapper == null) {
                LOGGER.warn("Mapper for type '{}' not found.", dataType);
                return Optional.empty();
            } else {
                if (TRIM_DATA_TYPES.contains(dataType)) {
                    defaultValueExpression = defaultValueExpression.trim();
                }

                try {
                    Object rawDefaultValue = mapper.parse(column, defaultValueExpression);
                    Object convertedDefaultValue =
                            this.convertDefaultValue(rawDefaultValue, column);
                    if (convertedDefaultValue == null) {
                        return Optional.empty();
                    } else if (convertedDefaultValue instanceof Struct) {
                        LOGGER.warn(
                                "Struct can't be used as default value for column '{}', will use null instead.",
                                column.name());
                        return Optional.empty();
                    } else {
                        return Optional.of(convertedDefaultValue);
                    }
                } catch (Exception var7) {
                    Exception e = var7;
                    LOGGER.warn(
                            "Cannot parse column default value '{}' to type '{}'. Expression evaluation is not supported.",
                            defaultValueExpression,
                            dataType);
                    LOGGER.debug("Parsing failed due to error", e);
                    return Optional.empty();
                }
            }
        }
    }

    private Object convertDefaultValue(Object defaultValue, Column column) {
        if (this.valueConverters != null && defaultValue != null) {
            SchemaBuilder schemaBuilder = this.valueConverters.schemaBuilder(column);
            if (schemaBuilder == null) {
                return defaultValue;
            } else {
                Schema schema = schemaBuilder.build();
                Field field = new Field(column.name(), -1, schema);
                ValueConverter valueConverter = this.valueConverters.converter(column, field);
                Object result = valueConverter.convert(defaultValue);
                if (result instanceof BigDecimal
                        && column.scale().isPresent()
                        && (Integer) column.scale().get() > ((BigDecimal) result).scale()) {
                    result =
                            ((BigDecimal) result)
                                    .setScale(
                                            (Integer) column.scale().get(), RoundingMode.HALF_EVEN);
                }

                return result;
            }
        } else {
            return defaultValue;
        }
    }

    private static DefaultValueConverter.DefaultValueMapper parseNullDefault(
            DefaultValueConverter.DefaultValueMapper mapper) {
        return (c, x) -> {
            return x.startsWith("NULL::") ? null : mapper.parse(c, x);
        };
    }

    private static Map<String, DefaultValueConverter.DefaultValueMapper> createDefaultValueMappers(
            TimestampUtils timestampUtils) {
        Map<String, DefaultValueConverter.DefaultValueMapper> result = new HashMap();
        result.put(
                "bit",
                (c, v) -> {
                    String defaultValue = extractDefault(v, "00");
                    return defaultValue.length() == 1 ? "1".equals(defaultValue) : defaultValue;
                });
        result.put(
                "varbit",
                (c, v) -> {
                    return extractDefault(v, "0");
                });
        result.put(
                "bool",
                parseNullDefault(
                        (c, v) -> {
                            return Boolean.parseBoolean(extractDefault(v, "false"));
                        }));
        result.put(
                "bpchar",
                (c, v) -> {
                    return extractDefault(v, "");
                });
        result.put(
                "varchar",
                (c, v) -> {
                    return extractDefault(v, "");
                });
        result.put(
                "text",
                (c, v) -> {
                    return extractDefault(v, "");
                });
        result.put(
                "numeric",
                parseNullDefault(
                        (c, v) -> {
                            return new BigDecimal(extractDefault(v, "0.0"));
                        }));
        result.put(
                "float4",
                parseNullDefault(
                        (c, v) -> {
                            return Float.parseFloat(extractDefault(v, "0.0"));
                        }));
        result.put(
                "float8",
                parseNullDefault(
                        (c, v) -> {
                            return Double.parseDouble(extractDefault(v, "0.0"));
                        }));
        result.put(
                "int2",
                parseNullDefault(
                        (c, v) -> {
                            return Short.parseShort(extractDefault(v, "0"));
                        }));
        result.put(
                "int4",
                parseNullDefault(
                        (c, v) -> {
                            return Integer.parseInt(extractDefault(v, "0"));
                        }));
        result.put(
                "serial",
                parseNullDefault(
                        (c, v) -> {
                            return Integer.parseInt(extractDefault(v, "0"));
                        }));
        result.put(
                "int8",
                parseNullDefault(
                        (c, v) -> {
                            return Long.parseLong(extractDefault(v, "0"));
                        }));
        result.put(
                "bigserial",
                parseNullDefault(
                        (c, v) -> {
                            return Long.parseLong(extractDefault(v, "0"));
                        }));
        result.put(
                "smallserial",
                parseNullDefault(
                        (c, v) -> {
                            return Short.parseShort(extractDefault(v, "0"));
                        }));
        result.put(
                "json",
                (c, v) -> {
                    return extractDefault(v, "{}");
                });
        result.put(
                "jsonb",
                (c, v) -> {
                    return extractDefault(v, "{}");
                });
        result.put(
                "xml",
                (c, v) -> {
                    return extractDefault(v, "");
                });
        result.put(
                "uuid",
                (c, v) -> {
                    return UUID.fromString(
                            extractDefault(v, "00000000-0000-0000-0000-000000000000"));
                });
        result.put(
                "date",
                (c, v) -> {
                    return timestampUtils.toLocalDateTime(extractDefault(v, "1970-01-01"));
                });
        result.put(
                "time",
                (c, v) -> {
                    return timestampUtils.toLocalTime(extractDefault(v, "00:00"));
                });
        result.put(
                "timestamp",
                (c, v) -> {
                    return timestampUtils.toOffsetDateTime(extractDefault(v, "1970-01-01"));
                });
        result.put(
                "timestamptz",
                (c, v) -> {
                    return timestampUtils
                            .toOffsetDateTime(extractDefault(v, "1970-01-01"))
                            .atZoneSameInstant(ZoneOffset.UTC);
                });
        result.put(
                "interval",
                (c, v) -> {
                    return new PGInterval(extractDefault(v, "epoch"));
                });
        return result;
    }

    private static String extractDefault(String defaultValue) {
        if (!defaultValue.contains("'")) {
            return defaultValue.startsWith("NULL::") ? null : defaultValue;
        } else {
            Matcher matcher = LITERAL_DEFAULT_PATTERN.matcher(defaultValue);
            matcher.find();
            return matcher.group(1);
        }
    }

    private static String extractDefault(String defaultValue, String generatedValuePlaceholder) {
        Matcher functionMatcher = FUNCTION_DEFAULT_PATTERN.matcher(defaultValue);
        return !functionMatcher.find() && !CURRENT_DATE_TIMES.contains(defaultValue.toLowerCase())
                ? extractDefault(defaultValue)
                : generatedValuePlaceholder;
    }

    public boolean supportConversion(String typeName) {
        return this.defaultValueMappers.containsKey(typeName);
    }
}
